package WindowFun;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

public class Book_WindowRes implements WindowFunction<Long, ClassObject.Book, Tuple, TimeWindow> {
    //定义一个时间格式化工具用来将当前时间(双十一那天订单的时间)转为String格式
    SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<Long> input, Collector<ClassObject.Book> out) throws Exception {
        String bookname = ((Tuple1<String>) tuple).f0;
        Long t = input.iterator().next();
        //统计销量数

        long currentTimeMillis = System.currentTimeMillis();
        String dateTime = df.format(currentTimeMillis);

        ClassObject.Book book = new ClassObject.Book(bookname,dateTime,t);
        out.collect(book);
    }
}
